mod performance;

use std::any::Any;
use std::time::{Duration, Instant};
use bincode::{Decode, Encode};
use log::info;
use rpc::{RpcChannel, RpcContext, RpcEvent, RpcFilter, RpcResult};
use rpc_client::client::RpcClientBuilder;
use rpc_server::server::RpcServerBuilder;
use rpc_derive::{router, rpc_router, rpc_service, service};
use serde::{Deserialize, Serialize};
use rpc::bootstrap::BootstrapBuilder;
use rpc::error::{CustomerErrorKind, RpcError};

#[tokio::test]
async fn server_test_json() -> RpcResult<()> {
    // env_logger::init();
    server_test(8881, "json", 1).await
}

#[tokio::test]
async fn server_test_json_100() -> RpcResult<()> {
    // env_logger::init();
    server_test(8882, "json", 100).await
}

#[tokio::test]
async fn server_test_bin_code() -> RpcResult<()> {
    env_logger::init();
    server_test(8883, "bin-code", 1).await
}

#[tokio::test]
async fn server_test_bin_code_1000() -> RpcResult<()> {
    server_test(8885,"bin-code", 10000).await
}

async fn server_test(port: u32, codec_type: &'static str, n: u32) -> RpcResult<()>{
    tokio::spawn(async move {
        let bootstrap = BootstrapBuilder::new()
            .router("/hello", router!(hello))
            .router("/world", router!(world))
            .filter("*", Box::new(AllFilter))
            .filter("/hello", Box::new(HelloFilter))
            ;
        let _ = RpcServerBuilder::new(bootstrap, port)
            .build().start_server().await;
    });

    let now = Instant::now();
    let bootstrap = BootstrapBuilder::new()
        .router("/echo", router!(echo))
        .build();
    let address = format!("127.0.0.1:{}", port);
    let client = RpcClientBuilder::new(bootstrap, &address)
        .set_current_codec_type(codec_type)
        .build();
    let mut channel = client.connection().await?;

    info!("client connection time: {:?}", now.elapsed());

    let mut hello_service = service!(channel.context.clone(), HelloService); //HelloServiceImpl{ context:
    // context.clone()}; //service!(&context, HelloService);
    let mut world_service = service!(channel.context.clone(), WorldService);

    for i in 0..n {
        let result = hello_service.hello(&mut channel, i as i32, "hello".to_string()).await;
        let result2 = world_service.world(&mut channel, "world".to_string()).await?;
        // info!("call result: {}, result2: {}", result, result2);
        match result {
            Ok(result) => {
                assert_eq!(result, (i+100) as i32);
            }
            Err(error) => {
                if let RpcError::CustomerError { message, ..} = error {
                    assert_eq!(message, "测试异常");
                }
            }
        }
        assert_eq!(result2, "world demo");
    }

    let time = now.elapsed();
    println!("result:::::::::::::::: time: {}", time.as_millis());

    tokio::time::sleep(Duration::from_millis(100)).await;
    Ok(())
}

#[rpc_router]
fn hello(context: &mut RpcContext, data: i32, _value: String) -> i32 {
    context.send_event(Box::new(HelloEvent {})).unwrap();

    let mut session = context.session();
    session.with_data("hello", |a: &mut i32| {
        println!("session value: {}", a);
        a.clone()
    });
    data + 100
}

#[rpc_router]
fn world(context: &mut RpcContext, value: String) -> String {
    context.send_event(Box::new(HelloEvent {})).unwrap();

    let mut result = value.clone();
    result.push_str(" demo");
    result
}

// pub struct HelloRouterFactory;
// #[derive(Debug)]
// pub struct HelloRouter {
//     pub data: i32,
//     pub value: String,
// }
//
// impl rpc::router::RouterFactory for HelloRouterFactory {
//     fn parse(&self, rpc_context: &dyn rpc::RpcContext, byte_muts: bytes::BytesMut) -> rpc::RpcResult<Box<dyn rpc::router::DoRouter>> {
//         let (data, value) = rpc::decode::<(i32, String)>(rpc_context, byte_muts.freeze().to_vec())?;
//         Ok(Box::new(HelloRouter {
//             data,
//             value,
//         }))
//     }
// }
// impl rpc::router::DoRouter for HelloRouter {
//     fn execute(&mut self, rpc_context: &mut dyn RpcContext) -> RpcResult<Vec<u8>> {
//         let result = hello(rpc_context, self.data.clone(), self.value.clone());
//         rpc::encode(rpc_context, result)
//     }
// }

#[rpc_service(timeout=30)]
#[async_trait::async_trait]
trait EchoService {

    #[path = "/echo"]
    async fn echo(&mut self, channel: &mut impl RpcChannel, value: i32) -> RpcResult<i32>;
}

#[derive(Debug)]
struct HelloEvent {}

impl RpcEvent for HelloEvent {
    fn consume(&self, context: RpcContext) -> RpcResult<()> {
        info!("received hello message");
        Ok(())
    }
}

#[rpc_router]
fn echo(_context: &mut RpcContext, d: EchoData) -> f32 {
    d.d1
}

#[derive(Debug, Serialize, Deserialize, Encode, Decode, Clone)]
struct EchoData {
    d1: f32,
    str: String,
}

#[rpc_service(timeout=30)]
#[async_trait::async_trait]
trait HelloService {

    #[path = "/hello"]
    async fn hello(&mut self, channel: &mut impl RpcChannel, value: i32, data: String) -> RpcResult<i32>;
}

#[rpc_service(timeout=30)]
#[async_trait::async_trait]
trait WorldService {

    // #[path(name="/hello", timeout=60)]
    #[path = "/world"]
    async fn world(&mut self, channel: &mut impl RpcChannel, data: String) -> RpcResult<String>;
}


// pub struct HelloServiceImpl<'a> {
//     context: &'a dyn rpc::RpcContext,
// }
// impl<'a> HelloServiceImpl<'a> {
//     pub fn new(context: &'a dyn rpc::RpcContext) -> HelloServiceImpl {
//
//         HelloServiceImpl {
//             context,
//         }
//     }
// }
// #[async_trait::async_trait]
// impl<'a> HelloService for HelloServiceImpl<'a> {
//     async fn hello(&mut self, channel: &mut impl RpcChannel, value: i32, data: String) -> RpcResult<i32> {
//         let bytes = rpc::encode(self.context, (value, data))?;
//         let local_seq = self.context.get_local_seq_creator().create_seq();
//         let header = rpc::dto::RpcHeader::new("/hello", local_seq.clone(), 0);
//         let rpc_data_bytes = rpc::dto::RpcDataBytes::new(header, bytes);
//         let result = rpc::encode_rpc_data_bytes(self.context, rpc_data_bytes)?;
//         let data = channel.write(local_seq, result, 30u32).await?;
//         let value = rpc::decode(self.context, data)?;
//         Ok(value)
//     }
// }


struct AllFilter;
impl RpcFilter for AllFilter {
    fn before(&self, _: &mut RpcContext, _: &mut dyn Any) -> RpcResult<bool> {
        info!("all filter before");
        Ok(true)
    }


    fn after(&self, _rpc_channel: &mut RpcContext, _data: &mut dyn Any) -> Result<(), RpcError> {
        info!("all filter after");
        Ok(())
    }
}

struct HelloFilter;
impl RpcFilter for HelloFilter {
    fn before(&self, context: &mut RpcContext, value: &mut dyn Any) -> RpcResult<bool> {
        let value = value.downcast_mut::<HelloRouter>().unwrap();
        value.data -= 10;
        if value.data.is_negative() {
            return Err(RpcError::CustomerError{error_kid: CustomerErrorKind::DataError, message: "测试异常".to_owned()});
        }
        let mut session = context.session();
        session.add_data("hello".to_string(), Box::new(value.data));
        info!("demo filter before, value: {}", value.data);
        Ok(true)
    }

    fn after(&self, context: &mut RpcContext, data: &mut dyn Any) -> Result<(), RpcError> {
        let value = data.downcast_mut::<i32>().unwrap();
        *value += 10;
        let mut session = context.session();
        let _t = session.remove_data("hello");
        info!("demo filter before, value: {}", value);
        Ok(())
    }
}

